package com.base.wslisten;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WSServerListen {
	
	private static long sessionId=0L;
	//监听套接字
	private final int port;
	
	private final static short MAX_THEARD_SUM=2;
	//安全消息队列 客户端缓存消息队列
	private static ConcurrentHashMap<Short, Queue<BaseClientSession>> reciveClientMessageQueue = new ConcurrentHashMap<>();
	private static ConcurrentHashMap<String, Long> allClientSessions = new ConcurrentHashMap<>();
	//这个是为了不把session暴露给gameserver
	private static ConcurrentHashMap<Long, String> allIdRelationSessionIdConcurrentHashMap =  new ConcurrentHashMap<>();
	public static ConcurrentHashMap<Long, String> getAllIdRelationSessionIdConcurrentHashMap() {
		return allIdRelationSessionIdConcurrentHashMap;
	}
	public static void setAllIdRelationSessionIdConcurrentHashMap(
			ConcurrentHashMap<Long, String> allIdRelationSessionIdConcurrentHashMap) {
		WSServerListen.allIdRelationSessionIdConcurrentHashMap = allIdRelationSessionIdConcurrentHashMap;
	}

	private final ServerBootstrap bootstrap;
	BaseRepeater baseRepeater;
	public WSServerListen(int port,BaseRepeater baseRepeater) {
			this.port = port;
			bootstrap = new ServerBootstrap();
			this.baseRepeater = baseRepeater;
		}
	//监听别的
	public void run() {
	    EventLoopGroup bossGroup = new NioEventLoopGroup();        // 用来接收进来的连接
	    EventLoopGroup workerGroup = new NioEventLoopGroup();    // 用来处理已经被接收的连接
	    System.out.println("准备运行端口：" + port);
	    
	    try {
	    	bootstrap.group(bossGroup, workerGroup)
	        .channel(NioServerSocketChannel.class)            // 这里告诉Channel如何接收新的连接
	        .childHandler( new ChannelInitializer<SocketChannel>() {
	            @Override
	            protected void initChannel(SocketChannel ch) {
	            	
	            	ChannelPipeline pipeline = ch.pipeline();	
	            	pipeline.addLast("http-codec",new HttpServerCodec());//设置解码器
	            	pipeline.addLast("aggregator",new HttpObjectAggregator(65536));//聚合器，使用websocket会用到
	            	pipeline.addLast("http-chunked",new ChunkedWriteHandler());//用于大数据的分区传输
	            	pipeline.addLast("handler",new WSServerListenHandler(port));//自定义的业务handler
	            }
	        })
	        .option(ChannelOption.SO_BACKLOG, 128)
	        .childOption(ChannelOption.SO_KEEPALIVE, true);
	        for(Short i =0; i<MAX_THEARD_SUM; i++) {
	        	final Short index = i;
				ExecutorService singleThreadExecutorService = Executors.newSingleThreadExecutor();
				singleThreadExecutorService.execute(() -> {
					// TODO Auto-generated method stub
					while(true){
						if(!reciveClientMessageQueue.isEmpty()) {
							Queue<BaseClientSession> clientSessions = reciveClientMessageQueue.get(index);
							if(clientSessions == null|| clientSessions.size() == 0) {
								try {
									Thread.sleep(500);
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
								continue;
							}
							BaseClientSession clientSession = clientSessions.poll();
							if(clientSession == null) {
								try {
									Thread.sleep(500);
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
								continue;
							}
							baseRepeater.sendMsg(clientSession);
						}else {
							 try {
								Thread.sleep(500);
							} catch (InterruptedException e) {
								e.printStackTrace();
							}
						}
					}
				});	
	        }
	        
        // 绑定端口，开始接收进来的连接
        ChannelFuture f = bootstrap.bind(port).sync(); 
        // 等待服务器socket关闭
        f.channel().closeFuture().sync();
	    } catch (Exception e) {
	        workerGroup.shutdownGracefully();
	        bossGroup.shutdownGracefully();
	    }
	}
	
	public static void reciveClientSession(ChannelHandlerContext ctx, String msgString) {
		String channleString = ctx.channel().id().asShortText();
		BaseClientSession session = null;
		if (allClientSessions.containsKey(channleString)) {

			Long id = allClientSessions.get(channleString);
			session = new BaseClientSession(ctx, msgString, id);
			Short threadPoolId = (short) (session.getUuidLong() % MAX_THEARD_SUM);
			Queue<BaseClientSession> clientSessions = reciveClientMessageQueue.get(threadPoolId);
			clientSessions.offer(session);

		} else {
			allClientSessions.put(channleString, ++sessionId);
			session = new BaseClientSession(ctx, msgString, sessionId);
			Short threadPoolId = (short) (session.getUuidLong() % MAX_THEARD_SUM);
			Queue<BaseClientSession> clientSessions = new LinkedList<>();
			clientSessions.offer(session);
			reciveClientMessageQueue.put(threadPoolId, clientSessions);
			allIdRelationSessionIdConcurrentHashMap.put(session.getUuidLong(), channleString);
		}
	}
}
